

namespace Stee.CAP8.MessageBus.Kafka.Factories
{
    using System;
    using Confluent.Kafka;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Factory to create new Kafka publisher
    /// </summary>
    public static class PublisherFactory
    {
        /// <summary>
        /// Create new publisher
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <param name="publisherErrorHandler">Publisher error handler</param>
        /// <returns>New publisher/producer instance</returns>
        public static IProducer<string, byte[]> CreateNewPublisher(string jsonConfig, Action<IProducer<string, byte[]>, Error> publisherErrorHandler)
        {
            var handler = publisherErrorHandler ?? throw new ArgumentNullException(nameof(publisherErrorHandler));

            var config = InitializePublisherConfig(jsonConfig);
            var client = BuildPublisher(config, handler);

            return client;
        }


        #region Private Methods

        /// <summary>
        /// Initialize publisher configuration
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <returns>Return new publisher/producer configuration</returns>
        private static ProducerConfig InitializePublisherConfig(string jsonConfig)
        {
            var configs = ConfigurationHelper.GetBrokerConfiguration(jsonConfig, ConfigurationConstants.BrokerConfigPublishSection);
            var configuration = new PublisherConfiguration(configs);

            return new ProducerConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                ClientId = configuration.ClientID,
                Acks = configuration.Acks,
                EnableIdempotence = configuration.EnableIdempotence,
                LingerMs = configuration.LingerMs,
                MessageSendMaxRetries = configuration.MessageSendMaxRetries,
                MessageTimeoutMs = configuration.MessageTimeoutMs,
                QueueBufferingMaxKbytes = configuration.QueueBufferingMaxKbytes,
                QueueBufferingMaxMessages = configuration.QueueBufferingMaxMessages,
                TransactionTimeoutMs = configuration.TransactionTimeoutMs,
            };
        }

        /// <summary>
        /// Build publisher
        /// </summary>
        /// <param name="producerConfig">Producer configuration</param>
        /// <param name="publisherErrorHandler">Producer error handler</param>
        /// <returns>Return new publisher/producer instance</returns>
        private static IProducer<string, byte[]> BuildPublisher(ProducerConfig producerConfig, Action<IProducer<string, byte[]>, Error> publisherErrorHandler)
        {
            var producerBuilder = new ProducerBuilder<string, byte[]>(producerConfig);
            producerBuilder.SetKeySerializer(Serializers.Utf8);
            producerBuilder.SetValueSerializer(Serializers.ByteArray);
            producerBuilder.SetErrorHandler(publisherErrorHandler);

            return producerBuilder.Build();
        }

        #endregion Private Methods
    }
}
